package org.lib.kafkahello.try2;

import java.util.Date;
import java.util.stream.IntStream;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;

/**
 * 试验2 的调试接口
 * @author ben
 * @date 2021-08-24 18:59:11 CST
 */
@RestController
@RequestMapping(value="/try2")
@Slf4j
public class Try2Controller {

	@Autowired
	private KafkaTemplate<String, Object> kafkaTemplate;
	
	/**
	 * 调用接口发送消息
	 * @author ben
	 * @date 2021-08-24 16:31:33 CST
	 * @return
	 */
	@GetMapping(value="/send")
	public Boolean sendMsg() {
		// 情况1、2、4测试：10条
		IntStream.range(0, 10).forEach(i->{
		// 情况3测试：20条
//		IntStream.range(0, 20).forEach(i->{
			log.info("sendMsg-{}", i);
			String msg = String.format("Try2 send msg-%d @%s", i, new Date());
			
			// 分别测试下面3种情况
			// 1.无key
//			kafkaTemplate.send(Try2Config.TOPIC_02, "no key-" + msg);
			
			// 2.都发送到了 分区0
			// 有key
//			kafkaTemplate.send(Try2Config.TOPIC_02, Try2Config.TOPIC_02_KEY, "fixed key-" + msg);
			
			// 3.发送消息条数改为 20条，key=i%4
			// 注意，第二个参数是 String类型！！
			kafkaTemplate.send(Try2Config.TOPIC_02, "" + i%4, "i%4 key-" + msg);
			
			// 4、指定分区发送
//			kafkaTemplate.send(Try2Config.TOPIC_02, i%Try2Config.TOPIC_02_P_NUMBER, "" + i%3, "fixed partition key-" + msg);
			// 错误：指定的 分区号 大于 分区数！
//			kafkaTemplate.send(Try2Config.TOPIC_02, i%3, "" + i%3, "fixed partition key-" + msg);
		});
		
		return true;
	}
	
}
